%%%-------------------------------------------------------------------
%%% File    : mapper_node_srv.erl
%%% Author  :  vjache
%%% Description : 
%%%
%%% Created : 15 Jan 2010 by  vjache
%%%-------------------------------------------------------------------
-module(mapper_node_srv).

-behaviour(gen_server).
%% --------------------------------------------------------------------
%% Include files
%% --------------------------------------------------------------------
-include_lib("eunit/include/eunit.hrl").
-include("../../include/erete.hrl").
-include("../../include/pipe.hrl").
-include("../../include/iter.hrl").

%% API
-export([start_link/1,
		 insert/2,
		 delete/2]).

%% gen_server callbacks
-export([init/1, 
		 handle_call/3, 
		 handle_cast/2, 
		 handle_info/2,
		 terminate/2, 
		 code_change/3]).

-record(state, {conf,beta_node_pipe}).

-record(mapper_node_conf,
		{id,
		 beta_node_id,
		 navi_srv_id,
		 map_fun
		}).
-define(IS_MAPPER_NODE_CONF_VALID(Conf),
		is_record(Conf,mapper_node_conf),
		is_atom(Conf#mapper_node_conf.id),
		is_atom(Conf#mapper_node_conf.beta_node_id),
		is_atom(Conf#mapper_node_conf.navi_srv_id),
		is_function(Conf#mapper_node_conf.map_fun,1)).

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
%% Description: Starts the server
%%--------------------------------------------------------------------
start_link(Conf) when 
  ?IS_MAPPER_NODE_CONF_VALID(Conf) ->
    gen_server:start_link(
	  {local, Conf#mapper_node_conf.id}, 
	  ?MODULE, 
	  [Conf], 
	  []).

insert(ServerRef,Value) ->
	server_call(ServerRef,{ins,Value}).
delete(ServerRef,Value) ->
	server_call(ServerRef,{del,Value}).

%%====================================================================
%% gen_server callbacks
%%====================================================================

%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, State} |
%%                         {ok, State, Timeout} |
%%                         ignore               |
%%                         {stop, Reason}
%% Description: Initiates the server
%%--------------------------------------------------------------------
init([Conf]) ->
    {ok, #state{
				conf=Conf,
				beta_node_pipe=pipe:open(beta_node_pipe(Conf))}
	}.

%%--------------------------------------------------------------------
%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
%%                                      {reply, Reply, State, Timeout} |
%%                                      {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, Reply, State} |
%%                                      {stop, Reason, State}
%% Description: Handling call messages
%%--------------------------------------------------------------------
handle_call(Request, From, State) ->
	try call(Request, From, State) of
		R -> R
	catch
		throw:Reason -> {reply, {error,Reason}, State}
	end.

%%--------------------------------------------------------------------
%% Function: handle_cast(Msg, State) -> {noreply, State} |
%%                                      {noreply, State, Timeout} |
%%                                      {stop, Reason, State}
%% Description: Handling cast messages
%%--------------------------------------------------------------------
handle_cast(_Msg, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: handle_info(Info, State) -> {noreply, State} |
%%                                       {noreply, State, Timeout} |
%%                                       {stop, Reason, State}
%% Description: Handling all non call/cast messages
%%--------------------------------------------------------------------
handle_info(_Info, State) ->
    {noreply, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, State) -> void()
%% Description: This function is called by a gen_server when it is about to
%% terminate. It should be the opposite of Module:init/1 and do any necessary
%% cleaning up. When it returns, the gen_server terminates with Reason.
%% The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _State) ->
    ok.

%%--------------------------------------------------------------------
%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
call(Msg={MdfCode,_Value}, _From, State) when 
  ?IS_MDF_CODE(MdfCode) ->
	BnPipeState=State#state.beta_node_pipe,
	BnPipeState1=pipe:push(BnPipeState,Msg),
	{reply, ok, State#state{beta_node_pipe=BnPipeState1}};
call(Request, From, State) ->
	throw({bad_call,{Request, From, State}}).

server_call(ServerRef,Request) ->
	case gen_server:call(ServerRef, Request, infinity) of
		{error,Reason} ->
			throw(Reason);
		Result ->
			Result
	end.

beta_node_pipe(#mapper_node_conf{
					  id=InpId,
					  beta_node_id=BnId,
					  navi_srv_id=_NavSrvId, %% TODO: use this param for failover
					  map_fun=MapFun
					 }) ->
	BnSrvSwitchPipe=pipe_switch:new(
				   fun({_MdfCode,_K,_V}) -> BnId end, 
				   fun(BnSrvRef)->
%% 						   pipe_mapper:new(
%% 							 fun(O)->
%% 									 {from,InpId,to,BnSrvRef,msg,O}
%% 							 end,pipe_console:new())
						   beta_node_rcv_srv:in_pipe(InpId, BnSrvRef)
				   end),
	MdfCodeDecor=fun(MdfCode,Iter)->
						 iter_mapper:new(Iter,fun({K,V})-> {MdfCode,K,V} end )
				 end,
	pipe_mapper:new(
	  fun({MdfCode,V}) when ?IS_MDF_CODE(MdfCode)->
			  case MapFun(V) of
				  Iter when ?IS_ITER(Iter) ->
					  MdfCodeDecor(MdfCode,Iter);
				  KVList when is_list(KVList) ->
					  MdfCodeDecor(MdfCode,
								   iter:from_list(KVList));
				  {K,V1} ->
					  {MdfCode,K,V1};
				  MapFunBadOut ->
					  throw({bad_map_fun_out,MapFunBadOut})
			  end;
		 (BadInput)->
			  throw({bad_input,BadInput})
	  end, 
	  pipe_iter:new(BnSrvSwitchPipe)).

%%--------------------------------------------------------------------
%%% eUnit tests
%%--------------------------------------------------------------------

pipe_test()->
	Pipe=beta_node_pipe(#mapper_node_conf{
					  id=inp_1,
					  beta_node_id=bn_1,
					  navi_srv_id=undefined, %% TODO: use this param for failover
					  map_fun=fun(Word)->
									  {erlang:phash2(Word),Word}
							  end
					 }),
	Iter=iter_mapper:new(iter:from_list(lists:seq(1, 100)),
						 fun(Item)->
								 {ins,Item}
						 end),
	Pump=pump:new(Iter, Pipe),
	pump:init(Pump).

srv_test()->
	start_link(#mapper_node_conf{
					  id=SrvRef=inp_1,
					  beta_node_id=bn_1,
					  navi_srv_id=undefined, %% TODO: use this param for failover
					  map_fun=fun(Word)->
									  {erlang:phash2(Word),Word}
							  end
					 }),
	[insert(SrvRef,I)||I<-lists:seq(1, 100)].	

full_test()->
	beta_node_sup:start_link(#beta_node_conf
		 {
		  id=BnId='bn#1', 								% an atom identifying beta_node server
		  in_ids=[Inp1='an#1',Inp2='an#2'],		 				% a list of atoms
		  out_pipe=pipe_console:new(), 					% an output pipe
		  beta_mem_fun=fun beta_memory_ets:open/1,		% beta memory factory fun -> #mh
		  red_fun=fun(_Modif,_K,IterList)->				% reduce function 
						  iter_join_cprod:all(IterList)
				  end,
		  red_mode = inc								% reduce mode (inc|aggr)
		 }),
	start_link(#mapper_node_conf{
					  id=Inp1,
					  beta_node_id=BnId,
					  navi_srv_id=undefined, %% TODO: use this param for failover
					  map_fun=fun({K,V})->
									  {K,V}
							  end
					 }),
	start_link(#mapper_node_conf{
					  id=Inp2,
					  beta_node_id=BnId,
					  navi_srv_id=undefined, %% TODO: use this param for failover
					  map_fun=fun({K,V})->
									  {K,V}
							  end
					 }),
	insert(Inp1,{1,"Vasya"}),
	insert(Inp2,{1,"Pupkin"}),
	receive
		after 7000 -> ok
	end,
	insert(Inp1,{1,"Vasiliy"}),
	ok.